package com.xuchao.rabbit.b4;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.xuchao.rabbit.utils.RabbitMqUtils;

import java.util.UUID;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

/**
 * @Author xuchao
 * @Date:2023/6/6 21:30
 * @description 发布确认
 * @Version 1.0
 */
/*
发布确认模式
    1.单个确认
    2.批量确认
    3.异步确认
 */
public class ConfirmMessage {
    //批量发消息的个数
    public static final int MESSAGE_COUNT = 1000;

    public static void main(String[] args) throws Exception {
        //1.单个确认
        long a = publicsh1();
        //2.批量确认
        long b = publicsh2();
        //3.异步确认
        long c = publish3();
        System.out.println(a);
        System.out.println(b);
        System.out.println(c);
    }

    //单个确认
    public static long publicsh1() throws Exception {
        Channel channel = RabbitMqUtils.getChannel();
        String queueName = UUID.randomUUID().toString();
//        channel.queueDeclare(queueName,false,false,false,null);
        //开启发布确认
        channel.confirmSelect();
        //开始时间
        long begin = System.currentTimeMillis();
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = i+"";
            channel.basicPublish("",queueName,null,message.getBytes());
            //单个消息就马上进行发布确认
            boolean flag = channel.waitForConfirms();
            if(flag){
                System.out.println("发送成功");
            }
        }
        //结束时间
        long end = System.currentTimeMillis();
        return end-begin;
    }

    //批量确认
    public static long publicsh2() throws Exception {
        Channel channel = RabbitMqUtils.getChannel();
        String queueName = UUID.randomUUID().toString();
//        channel.queueDeclare(queueName,false,false,false,null);
        //开启发布确认
        channel.confirmSelect();
        //批量确认的大小
        int batchSize = 100;
        //开始时间
        long begin = System.currentTimeMillis();
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = i+"";
            channel.basicPublish("",queueName,null,message.getBytes());
            //批量确认
            if (i % batchSize == 0){
                boolean flag = channel.waitForConfirms();
                if(flag){
                    System.out.println("发送成功");
                }
            }
        }
        //结束时间
        long end = System.currentTimeMillis();
        return end-begin;
    }

    //异步发布确认
    public static long publish3() throws Exception {
        Channel channel = RabbitMqUtils.getChannel();
        String queueName = UUID.randomUUID().toString();
//        channel.queueDeclare(queueName,false,false,false,null);
        //开启发布确认
        channel.confirmSelect();

        //线程安全有序的一个哈希表，适用于高并发的情况下
        //1.轻松的将序号与消息进行关联
        //2.可以轻松的批量删除条目 只要给到序号
        //3.支持高并发
        ConcurrentSkipListMap<Long,String> outstandingConfirms = new ConcurrentSkipListMap<>();
        //开始时间
        long begin = System.currentTimeMillis();

        //消息确认成功 回调函数
        ConfirmCallback ackCallback = (deliveryTag,multiple)->{
            if(multiple){
                ConcurrentNavigableMap<Long, String> confirmed = outstandingConfirms.headMap(deliveryTag);
                confirmed.clear();;
            }else{
                outstandingConfirms.remove(deliveryTag);
            }
//            System.out.println("确认的消息："+deliveryTag);
        };
        //消息确认失败 回调函数
        ConfirmCallback nackCallback = (deliveryTag,multiple)->{
//            System.out.println("未确认消息："+deliveryTag);
        };
        //添加消息确认监听器
        channel.addConfirmListener(ackCallback,nackCallback);
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = i+"";
            channel.basicPublish("",queueName,null,message.getBytes());
            //1.此处记录下所有要发送的消息 消息的总和
            outstandingConfirms.put(channel.getNextPublishSeqNo(),message);
        }
        //结束时间
        long end = System.currentTimeMillis();
        return end-begin;
    }
}
